/*
 *  Copyright 2013 National Institute of Advanced Industrial Science and Technology
 *  
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *  
 *      http://www.apache.org/licenses/LICENSE-2.0
 *  
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 */

package org.sss.common.io;

import java.util.List;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.sss.common.CanceledException;
import org.sss.mapreduce.Configuration;
import org.sss.mapreduce.SssException;
import org.sss.mapreduce.GroupID;

/**
 * .
 * NOTE: access int only this module.
 */
abstract class AbstractDataReader implements DataReader {
  @SuppressWarnings("unused")
  private static Logger logger = LoggerFactory.getLogger(AbstractDataReader.class);

  private static class Counters {
    private long inputCount = 0L;

    public void countup(List<byte[]> kvs) {
      inputCount += kvs.size() / 2;
    }

    public void countup(int n, int nbytes) {
      inputCount += n;
    }

    public long getInputCount() {
      return inputCount;
    }

    public long getInputNBytes() {
      return 0L;
    }
  }

  private static class CountersWithBytes extends Counters {
    private long inputNBytes = 0L;

    public void countup(List<byte[]> kvs) {
      super.countup(kvs);
      for (byte[] item: kvs) {
        inputNBytes += item.length;
      }
    }

    public void countup(int n, int nbytes) {
      super.countup(n, nbytes);
      inputNBytes += nbytes;
    }

    public long getInputNBytes() {
      return inputNBytes;
    }
  }

  private boolean isCanceled = false;
  private Counters counters;
  private GroupID source;

  public AbstractDataReader(GroupID source, Configuration conf) throws SssException {
    this.source = source;
    if (conf.getBoolean("count.input.size", true)) {
      this.counters = new CountersWithBytes();
    } else {
      this.counters = new Counters();
    }
  }

  @Override
  public synchronized void cancel() {
    isCanceled = true;
  }

  protected synchronized void checkCanceled() throws SssException {
    if (isCanceled) {
      throw new CanceledException("Request is canceled.");
    }
  }

  @Override
  public long getInputCount() {
    return this.counters.getInputCount();
  }

  @Override
  public long getInputNBytes() {
    return this.counters.getInputNBytes();
  }

  protected void countup(List<byte[]> kvs) {
    counters.countup(kvs);
  }

  protected void countup(int n, int bytes) {
    counters.countup(n, bytes);
  }

  @Override
  public GroupID getSource() {
    return source;
  }

  protected boolean isDestructiveRead() {
    return source.isDestructive();
  }
}
